#include "ae.h"
#include "ae_epoll.h"
#include "connection.h"
#include "server.h"
#include "iothreadpool.h"

eventLoop::eventLoop(int setsize)
:setsize_(setsize)
{
    ets_ = tcallocArray(fileEvent, setsize);
    fire_ets_ = tcallocArray(fireEvent, setsize);

    state_ = Epoll::EpollCreate(setsize);

    for(int i = 0; i < setsize_; ++i)
    {
        ets_[i].mask_ = AE_NONE;
    }
}

//执行connection中的读写回调函数
//TODO clientdata怎么传
void eventLoop::connSocketEventHandler(int fd, void* clientdata, int mask)
{
    connection* conn = (connection*)clientdata;

    int call_write = (mask & AE_WRITABLE) && conn->write_func_;
    int call_read = (mask & AE_READABLE) && conn->read_func_;

    if(call_write) conn->write_func_(conn);
    if(call_read) conn->read_func_(conn);
}

//直接加到eventLoop的evnet中，没有加到conn的accept_handle
void eventLoop::acceptEvent(int fd, void* connData, int mask)
{
    struct sockaddr_in peer;
    socklen_t len = sizeof(peer);
    int clientfd = connection::Accept(fd, &peer, &len);
    if(clientfd < 0)
    {
        return;
    }
    //todo 需要new conn 加到全局队列中
    connection* conn = tcalloc(connection);
    conn->fd_ = clientfd;
    conn->read_func_ = IOThreadPool::readQueryFromClient;
    conn->endpoint_ = tcalloc(HttpEndPoint, conn);
    std::cout << "get a link..." << clientfd << "conn: " << conn << std::endl;

    conn->setNoBlock();
    //todo，写一个执行ConnectionCallbackFunc回调函数的代码
    server.el_->createFileEvent(clientfd, AE_READABLE, connSocketEventHandler, conn);
}

int eventLoop::createFileEvent(int fd, int mask, fileProc func, void* connData)
{
    if(fd >= setsize_)
        return -1;
    
    struct fileEvent* et = &ets_[fd];

    if(Epoll::AddEpoll(this, fd, mask) < 0)
        return -1;

    et->conn_data_ = connData;
    et->mask_ |= mask;
    if(mask & AE_READABLE) et->rfileproc_ = func;
    if(mask & AE_WRITABLE) et->wfileproc_ = func;
    return 0;
}

int eventLoop::deleteFileEvent(int fd, int mask)
{
    //TODO
    if(fd >= setsize_ || fd < 0)
        return -1;
    
    fileEvent* fe = &ets_[fd];
    if(Epoll::DelEpoll(this, fd, mask) < 0)
        return -1;
    fe->mask_ &= (~mask);

    return 0;
}

int eventLoop::processEvents()
{
    server.handleClientsWithPendingReadsUsingThreads();

    server.handleClientsWithPendingWritesUsingThreads();

    server.freeClientsInAsyncFreeQueue();

    int num = Epoll::WaitEpoll(this, -1);
    for(int i = 0; i < num; ++i)
    {
        int mask = fire_ets_[i].mask_;
        int fd = fire_ets_[i].fd_;
        struct fileEvent* fe = &ets_[fd];
        int fired = 0;

        if(mask & fe->mask_ & AE_READABLE)
        {
            fe->rfileproc_(fd, fe->conn_data_, mask);
            fired++;
        }
        
        if(mask & fe->mask_ & AE_WRITABLE)
        {
            //是否执行过
            if(!fired || fe->wfileproc_ != fe->rfileproc_)
                fe->wfileproc_(fd, fe->conn_data_, mask);
        }
    }
}